package com.wx.spider.pc.listener;

import com.wx.spider.rocketmq.annotation.WsRocketMQConsumer;
import com.wx.spider.rocketmq.consumer.BaseRocketMQConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

/**
 * @author:feng
 * @create:2021-04-23 10:54
 * 测试封装的广播消费
 */
//@Component
@Slf4j
@WsRocketMQConsumer(consumerGroup = "ws-consumer-pm-to-java-socket", topic = "ws-get-list-socket", consumeThreadMin = 4, messageModel = MessageModel.BROADCASTING)
public class PMToJavaSocketConsumer extends BaseRocketMQConsumer {
    @Override
    protected ConsumeConcurrentlyStatus consume(String tag, String topic, MessageExt messageExt) {
        int count = 0;
        try {

            //获取重试次数
            count = messageExt.getReconsumeTimes();
            //这里设置重试大于3次 那么通过保存数据库 人工来兜底
            if (count > 3) {
                log.info("该消息已经重试3次,保存数据库。topic={},keys={},msg={}", messageExt.getTopic(), messageExt.getKeys(), messageExt.getBody());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            log.info("topic:" + topic + "tags:" + tag);
            Thread.sleep(3000);
            String msg = new String(messageExt.getBody());
            log.info(messageExt.getTopic() + ":" + messageExt.getTags() + "-" + messageExt.getKeys() + "--> 消息:" + msg);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            e.printStackTrace();
            log.info("消费失败，当前消费重试次数为 = {}", count);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
